package mylab.starters.mq.impl.local;


import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.Subscribe;
import mylab.starters.mq.api.Message;
import mylab.starters.mq.api.MessageListener;
import mylab.starters.mq.api.Properties;
import mylab.starters.mq.impl.AbstractConsumer;
import mylab.utils.common.guava.AsyncEventUtil;

import java.util.Optional;

public class LocalConsumer extends AbstractConsumer {

    private GuavaListener listener;

    public LocalConsumer(Properties properties, String groupId, String clientId) {
        super(properties, groupId, clientId);
    }

    @Override
    protected void onSubscribe(String topic, MessageListener messageListener, String... tags) throws Exception {
        listener = new GuavaListener(topic, messageListener);
        AsyncEventUtil.regist(listener);
    }

    @Override
    protected void onUnsubscribe() throws Exception {
        AsyncEventUtil.unregist(listener);
    }

    static class GuavaListener {
        String topic;
        MessageListener messageListener;

        public GuavaListener(String topic, MessageListener messageListener) {
            this.topic = topic;
            this.messageListener = messageListener;
        }

        @Subscribe
        @AllowConcurrentEvents
        public void onEvent(Message message) {
            Optional.ofNullable(messageListener).ifPresent(l -> l.onSuccess(message));
        }
    }
}
